"""
Case Type   : mot
Case Name   : 打开enable_mot_server参数，使用gs_dump/gs_restore导出恢复，前后数据一致
Create At   : 2025/4/8
Owner       : @li-xin12345
Description : (编译带mot，不编译ENABLE_X86_RDTSCP，对应转测包)
    1、设置enable_incremental_checkpoint=off，enable_mot_server=on
    2、建表
    3、查看表占用磁盘大小、数据量等
    4、导出数据
    5、删除表
    6、恢复数据
    7、查看表占用磁盘大小、数据量等，与备份前对比
    8、清理环境
Expect      :
    1、成功
    2、成功
    3、成功
    4、成功
    5、成功
    6、成功
    7、一致
    8、成功
History     :
"""

import os
import time
import unittest

from yat.test import Node
from yat.test import macro

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger


class MotTest(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.pri_db_sh = CommonSH('PrimaryDbUser')
        self.pri_node_sh = Node('PrimaryDbUser')
        self.constant = Constant()
        self.common = Common()
        self.tb_name = 't_guc_mot_0028'
        self.view_name = 'v_guc_mot_0028'
        self.dump_file = 'f_guc_mot_0028'
        self.param_names = ['enable_incremental_checkpoint',
                            'enable_mot_server']
        self.default_values = [self.common.show_param(i) for i in
                               self.param_names]
        self.dump_path = os.path.join(os.path.dirname(
            macro.DB_INSTANCE_PATH), self.dump_file)

    def test_mot(self):
        text = '-----step1:设置enable_incremental_checkpoint=off，' \
               'enable_mot_server=on;expect:成功-----'
        self.log.info(text)
        res = self.pri_db_sh.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        f'{self.param_names[0]}=off')
        self.log.info(res)
        self.assertTrue(res, "执行gs_guc set失败，预期成功但未找到标志信息")
        res = self.pri_db_sh.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        f'{self.param_names[1]}=on')
        self.log.info(res)
        self.assertTrue(res, "执行gs_guc set失败，预期成功但未找到标志信息")
        res = self.pri_db_sh.restart_db_cluster()
        self.log.info(res)
        self.assertTrue(res, '执行失败：' + text)
        status = self.pri_db_sh.get_db_cluster_status()
        self.assertTrue("Normal" or "Degrade" in status)
        res = [self.common.show_param(i) for i in self.param_names]
        self.log.info(res)
        self.assertEqual(['off', 'on'], res, '执行失败：' + text)

        text = '-----step2:建表;expect:成功-----'
        self.log.info(text)
        sql = f'''drop foreign table if exists {self.tb_name} cascade;
        create foreign table {self.tb_name} (id int primary key, 
        name varchar(100) not null, age int not null default 0);
        create index idx_name on {self.tb_name} (name);
        insert into {self.tb_name} (id, name, age) values (1, 'alice', 25), 
        (2, 'bob', 30),(3, 'charlie', 35);
        select * from {self.tb_name} where age > 30;
        update {self.tb_name} set age = 26 where id = 1;
        begin;
        insert into {self.tb_name} (id, name, age) values (4, 'david', 40);
        commit;
        create or replace view {self.view_name} as
        select id, name, age from {self.tb_name} where age > 30;
        select * from {self.view_name};
        '''
        sql_res = self.pri_db_sh.execut_db_sql(sql)
        self.log.info(sql_res)
        self.assertIn(self.constant.DROP_FOREIGN_SUCCESS_MSG, sql_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.CREATE_FOREIGN_SUCCESS_MSG, sql_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.CREATE_INDEX_SUCCESS_MSG, sql_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, sql_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.COMMIT_SUCCESS_MSG, sql_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.CREATE_VIEW_SUCCESS_MSG, sql_res,
                      '执行失败：' + text)
        self.assertIn('2 rows', sql_res, '执行失败：' + text)

        text = '-----step3:查看表占用磁盘大小、数据量等-----'
        self.log.info(text)
        sql = f"select pg_size_pretty(pg_total_relation_size(" \
              f"'{self.tb_name}')) as table_size;"
        res = self.pri_db_sh.execut_db_sql(sql)
        self.log.info(res)
        self.assertNotIn('ERROR', res, '执行失败：' + text)

        sql = f'select checksum({self.tb_name}::text) from {self.tb_name};'
        before_size = self.pri_db_sh.execut_db_sql(sql)
        self.log.info(before_size)
        self.assertNotIn('ERROR', before_size, '执行失败：' + text)

        self.log.info('-----step4:导出数据-----')
        self.log.info('---导出---')
        dump_cmd = f"source {macro.DB_ENV_PATH};" \
                   f"gs_dump " \
                   f"-p {self.pri_node_sh.db_port} " \
                   f"{self.pri_node_sh.db_name} -F c " \
                   f"-f {self.dump_path}"
        self.log.info(dump_cmd)
        dump_msg = self.pri_node_sh.sh(dump_cmd).result()
        self.log.info(dump_msg)
        self.assertIn(self.constant.GS_DUMP_SUCCESS_MSG, dump_msg,
                      '执行失败:' + text)

        self.log.info('-----step5:删除表;expect:成功-----')
        sql = f"drop foreign table if exists {self.tb_name} cascade;checkpoint;"
        self.log.info(sql)
        res = self.pri_db_sh.execut_db_sql(sql)
        self.log.info(res)
        self.assertIn(self.constant.DROP_FOREIGN_SUCCESS_MSG, res,
                      '执行失败：' + text)
        status = self.pri_db_sh.get_db_cluster_status(param='status')
        self.log.info(status)
        if status is True:
            self.log.info('--等待集群数据同步--')
            flag = self.pri_db_sh.check_data_consistency()
            self.assertTrue(flag, '执行失败：' + text)

        self.log.info('-----step6:恢复数据;expect:成功-----')
        cmd = f"source {macro.DB_ENV_PATH};" \
              f"gs_restore " \
              f"-p {self.pri_node_sh.db_port} " \
              f"-d {self.pri_node_sh.db_name} " \
              f"{self.dump_path}"
        self.log.info(cmd)
        res = self.pri_node_sh.sh(cmd).result()
        self.log.info(res)
        self.assertIn(self.constant.RESTORE_SUCCESS_MSG, res,
                      '执行失败:' + text)

        text = ('-----step7:再次查看表占用磁盘大小、数据量等，与备份前对比;'
                'expect:一致-----')
        self.log.info(text)
        sql = f"select pg_size_pretty(pg_total_relation_size(" \
              f"'{self.tb_name}')) as table_size;"
        res = self.pri_db_sh.execut_db_sql(sql)
        self.log.info(res)
        self.assertNotIn('ERROR', res, '执行失败：' + text)

        sql = f'select checksum({self.tb_name}::text) from {self.tb_name};'
        after_size = self.pri_db_sh.execut_db_sql(sql)
        self.log.info(after_size)
        self.assertNotIn('ERROR', before_size, '执行失败：' + text)
        self.assertEqual(before_size, after_size, '执行失败：' + text)

    def tearDown(self):
        text = '-----step8:清理环境; expect:成功-----'
        self.log.info(text)
        cmd = f'rm -rf {self.dump_path};' \
              f'if [ -d {self.dump_path} ]; ' \
              f'then echo "exists"; else echo "not exists"; fi'
        rm_res = self.common.get_sh_result(self.pri_node_sh, cmd)

        sql = f'drop foreign table if exists {self.tb_name} cascade;checkpoint;'
        drop_res = self.pri_db_sh.execut_db_sql(sql)
        self.log.info(drop_res)
        time.sleep(10)

        res = [self.common.show_param(i) for i in self.param_names]
        self.log.info(res)
        if res != self.default_values:
            for i, param_name in enumerate(self.param_names):
                res = self.pri_db_sh.execute_gsguc(
                    'set', self.constant.GSGUC_SUCCESS_MSG,
                    f'{param_name}={self.default_values[i]}')
                self.log.info(res)
            restart_res = self.pri_db_sh.restart_db_cluster()
            self.log.info(restart_res)
            status = self.pri_db_sh.get_db_cluster_status()
            self.log.info(status)
            show_param = [self.common.show_param(i) for i in self.param_names]
            self.log.info(show_param)
            self.assertEqual(self.default_values, show_param,
                             '执行失败：' + text)

        self.assertEqual('not exists', rm_res, '文件删除失败')
        self.assertIn(self.constant.DROP_FOREIGN_SUCCESS_MSG, drop_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.CHECKPOINT_SUCCESS_MSG, drop_res,
                      '执行失败：' + text)
        self.log.info(f'-----{os.path.basename(__file__)} end-----')